package com.shop.cereshop.app.tencent.im.delay.service;

import com.shop.cereshop.app.tencent.im.delay.handler.DelayedTaskHandler;
import com.shop.cereshop.app.tencent.im.delay.model.DelayedTask;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @description:  抽象Redisson延迟队列服务（支持多任务）
 * @author zdd
 * @date 2025/9/24 10:26
 * @version 1.0
 */
@Service
@Slf4j
public class RDelayedQueueService {

    @Resource
    private RedissonClient redissonClient;

    // 按业务类型存储的延迟队列和目标队列
    // 延迟队列 是“投递口”，往这里放带延迟时间的消息，然后消费者线程会按时间顺序取出消息并处理
    private final Map<String, RDelayedQueue<DelayedTask>> delayedQueues = new ConcurrentHashMap<>();
    // 目标队列 是“到期收件箱”，消息到期后自动转移到这里，消费者线程从这里取出消息并处理
    private final Map<String, RBlockingQueue<DelayedTask>> destinationQueues = new ConcurrentHashMap<>();
    private final Map<String, ExecutorService> consumerExecutors = new ConcurrentHashMap<>();

    private volatile boolean running = true;

    // 任务处理器映射（按 taskType 注册）
    private final Map<String, DelayedTaskHandler> handlers = new ConcurrentHashMap<>();

    /**
     * 初始化指定业务类型的队列
     */
    public void initQueue(String businessType) {
//        String destinationName = "delay_queue_destination_" + businessType;
//        RBlockingQueue<DelayedTask> destinationQueue = redissonClient.getBlockingQueue(destinationName);
        RBlockingQueue<DelayedTask> destinationQueue = redissonClient.getBlockingQueue(businessType);
        RDelayedQueue<DelayedTask> delayedQueue = redissonClient.getDelayedQueue(destinationQueue);

        destinationQueues.put(businessType, destinationQueue);
        delayedQueues.put(businessType, delayedQueue);

        // 启动该业务类型的消费者线程
        startConsumer(businessType, destinationQueue);
        log.info("Redisson延迟队列服务初始化完成，业务类型: {}", businessType);
    }

    /**
     * 添加延迟任务
     * @param businessType 业务类型,例如：red_packet_expire_refund、transfer_order_expire_refund等
     * @param task 延迟任务
     * @param delaySeconds 延迟时间（秒）
     */
    public void addTask(String businessType, DelayedTask task, long delaySeconds) {
        try {
            RDelayedQueue<DelayedTask> queue = delayedQueues.get(businessType);
            if (queue == null) {
                throw new RuntimeException("未初始化的业务队列: " + businessType);
            }
            queue.offer(task, delaySeconds, TimeUnit.SECONDS);
            log.info("任务 taskId: [{}] 类型: [{}] 已加入 {} 延迟队列", task.getTaskId(), task.getTaskType(), businessType);
        } catch (Exception e) {
            log.error("添加延迟任务失败: {}", task.getTaskId(), e);
            throw new RuntimeException("添加延迟任务失败", e);
        }
    }

    public void registerHandler(DelayedTaskHandler handler) {
        handlers.put(handler.getTaskType(), handler);
        log.info("注册延迟任务处理器: {}", handler.getTaskType());
    }

    private void startConsumer(String businessType, RBlockingQueue<DelayedTask> destinationQueue) {
        ExecutorService executor = Executors.newSingleThreadExecutor(r -> {
            Thread t = new Thread(r, "Delay-Consumer-" + businessType);
            t.setDaemon(false);
            return t;
        });

        consumerExecutors.put(businessType, executor);

        executor.submit(() -> {
            log.info("{} 延迟消费者线程启动...", businessType);
            while (running) {
                try {
                    // 阻塞等待消息,直到有消息
                    DelayedTask task = destinationQueue.poll(10, TimeUnit.SECONDS);
                    if (task == null) {
                        // 超时无消息，继续等待
                        continue;
                    }

                    DelayedTaskHandler handler = handlers.get(task.getTaskType());
                    if (handler == null) {
                        log.warn("未找到任务处理器: {}", task.getTaskType());
                        continue;
                    }

                    boolean success = handler.handle(task);
                    if (success) {
                        log.info("延迟任务处理成功: taskId={}, type={}", task.getTaskId(), task.getTaskType());
                    } else {
                        log.error("延迟任务处理失败: taskId={}, type={}", task.getTaskId(), task.getTaskType());
                    }

                } catch (InterruptedException e) {
                    log.warn("消费者线程被中断");
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("处理延迟任务异常", e);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ex) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
            log.info("{} 延迟消费者线程已停止", businessType);
        });
    }

    @PreDestroy
    public void destroy() {
        running = false;

        // 关闭所有消费者线程
        for (Map.Entry<String, ExecutorService> entry : consumerExecutors.entrySet()) {
            ExecutorService executor = entry.getValue();
            // 先尝试优雅关闭
            executor.shutdown();
            try {
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    // 30秒后强制关闭
                    executor.shutdownNow();
                }
            } catch (InterruptedException e) {
                executor.shutdownNow();
            }
        }

        // 销毁所有队列
        for (RDelayedQueue<DelayedTask> queue : delayedQueues.values()) {
            queue.destroy();
        }

        log.info("Redisson通用延迟队列服务已关闭");
    }


}
